/*
 * File      : pipe.c
 * This file is part of RT-Thread RTOS
 * COPYRIGHT (C) 2012, RT-Thread Development Team
 *
 * The license and distribution terms for this file may be
 * found in the file LICENSE in this distribution or at
 * http://www.rt-thread.org/license/LICENSE
 *
 * Change Logs:
 * Date           Author       Notes
 * 2012-09-30     Bernard      first version.
 */
#include <rthw.h>
#include <rtthread.h>
#include <rtdevice.h>

static rt_size_t rt_pipe_read(rt_device_t dev, rt_off_t pos, void *buffer, rt_size_t size)
{
	rt_uint32_t level;
	rt_thread_t thread;
	struct rt_pipe_device *pipe;
	rt_size_t read_nbytes;

	pipe = PIPE_DEVICE(dev);
	RT_ASSERT(pipe != RT_NULL);

	thread = rt_thread_self();

	/* current context checking */
	RT_DEBUG_NOT_IN_INTERRUPT;

	do 
	{
		level = rt_hw_interrupt_disable();
		read_nbytes = rt_ringbuffer_get(&(pipe->ringbuffer), buffer, size);
		if (read_nbytes == 0)
		{
			rt_thread_suspend(thread);
			/* waiting on suspended read list */
			rt_list_insert_before(&(pipe->suspended_read_list), &(thread->tlist));
			rt_hw_interrupt_enable(level);

			rt_schedule();
		}
		else
		{
			if (!rt_list_isempty(&pipe->suspended_write_list))
			{
				/* get suspended thread */
				thread = rt_list_entry(pipe->suspended_write_list.next, 
					struct rt_thread, tlist);

				/* resume the write thread */
				rt_thread_resume(thread);
				rt_hw_interrupt_enable(level);

				rt_schedule();
			}
			else
			{
				rt_hw_interrupt_enable(level);
			}
			break;
		}
	} while (read_nbytes == 0);

	return read_nbytes;
}

struct rt_pipe_device *_pipe = RT_NULL;
static rt_size_t rt_pipe_write(rt_device_t dev, rt_off_t pos, const void *buffer, rt_size_t size)
{
	rt_uint32_t level;
	rt_thread_t thread;
	struct rt_pipe_device *pipe;
	rt_size_t write_nbytes;

	pipe = PIPE_DEVICE(dev);
	RT_ASSERT(pipe != RT_NULL);
	if (_pipe == RT_NULL)
		_pipe = pipe;
	
	thread = rt_thread_self();

	/* current context checking */
	RT_DEBUG_NOT_IN_INTERRUPT;

	do 
	{
		level = rt_hw_interrupt_disable();
		write_nbytes = rt_ringbuffer_put(&(pipe->ringbuffer), buffer, size);
		if (write_nbytes == 0)
		{
			/* pipe full, waiting on suspended write list */
			rt_thread_suspend(thread);
			/* waiting on suspended read list */
			rt_list_insert_before(&(pipe->suspended_write_list), &(thread->tlist));
			rt_hw_interrupt_enable(level);

			rt_schedule();
		}
		else
		{
			if (!rt_list_isempty(&pipe->suspended_read_list))
			{
				/* get suspended thread */
				thread = rt_list_entry(pipe->suspended_read_list.next, 
					struct rt_thread, tlist);

				/* resume the read thread */
				rt_thread_resume(thread);
				rt_hw_interrupt_enable(level);

				rt_schedule();
			}
			else
			{
				rt_hw_interrupt_enable(level);
			}
			break;
		}
	}while (write_nbytes == 0);

	return write_nbytes;
}

static rt_err_t rt_pipe_control(rt_device_t dev, rt_uint8_t cmd, void *args)
{
	return RT_EOK;
}

rt_err_t rt_pipe_create(const char* name, rt_size_t size)
{
	rt_err_t result = RT_EOK;
	rt_uint8_t* rb_memptr = RT_NULL;
	struct rt_pipe_device *pipe = RT_NULL;

	/* get aligned size */
	size = RT_ALIGN(size, RT_ALIGN_SIZE);
	pipe = (struct rt_pipe_device*) rt_calloc (1, sizeof(struct rt_pipe_device));
	if (pipe != RT_NULL)
	{
		/* create ring buffer of pipe */
		rb_memptr = rt_malloc(size);
		if (rb_memptr == RT_NULL)
		{
			result = -RT_ENOMEM;
			goto __exit;
		}
		/* initialize suspended list */
		rt_list_init(&pipe->suspended_read_list);
		rt_list_init(&pipe->suspended_write_list);
		
		/* initialize ring buffer */
		rt_ringbuffer_init(&pipe->ringbuffer, rb_memptr, size);

		/* create device */
		pipe->parent.type 	= RT_Device_Class_Char;
		pipe->parent.init 	= RT_NULL;
		pipe->parent.open 	= RT_NULL;
		pipe->parent.close 	= RT_NULL;
		pipe->parent.read 	= rt_pipe_read;
		pipe->parent.write 	= rt_pipe_write;
		pipe->parent.control = rt_pipe_control;

		return rt_device_register(&(pipe->parent), name, RT_DEVICE_FLAG_RDWR);
	}
	else
	{
		result = -RT_ENOMEM;
	}

__exit:
	if (pipe != RT_NULL) rt_free(pipe);
	if (rb_memptr != RT_NULL) rt_free(rb_memptr);

	return result;
}
RTM_EXPORT(rt_pipe_create);

void rt_pipe_destroy(struct rt_pipe_device* pipe)
{
	if (pipe == RT_NULL) return;

	/* un-register pipe device */
	rt_device_unregister(&(pipe->parent));

	/* release memory */
	rt_free(pipe->ringbuffer.buffer_ptr);
	rt_free(pipe);

	return;
}
RTM_EXPORT(rt_pipe_destroy);

